#![allow(unused_variables)]
#![allow(unused_imports)]
#![allow(unreachable_code)]
#![allow(dead_code)]
#![allow(unused_must_use)]
use warp::Filter ;

use std::io::Read ;
use async_std::fs ;

use rust_embed::RustEmbed;

#[derive(RustEmbed)]
#[folder = "../../data-model/schema/"]
struct Asset;

#[tokio::main]
async fn main() {
    println!("start server at http://0.0.0.0:3031/logging") ;

    let cors = warp::cors()
        .allow_any_origin()
        .allow_credentials(true)
        .allow_headers(vec!["Content-Type", "Authorization", "User-Agent", "Sec-Fetch-Mode", "Referer", "Origin", "Access-Control-Request-Method", "Access-Control-Request-Headers"])
        .allow_methods(vec!["GET", "POST", "DELETE", "OPTIONS"]);
    warp::serve( warp::path("logging").and(filters::routes()).with(cors) )
        .run(([0,0,0,0], 3031))
        .await ;
}

mod filters {
    use warp::Filter;
    use super::handlers ;

    pub fn routes() -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        (warp::path("ping").and(warp::get()).map(|| "pong"))
            .or(avro_buffer_echo())
            .or(json_buffer_echo())
            .or(avro_buffer())

    }
    pub fn json_buffer_echo() -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path!("sys" / String / "bucket" / String / "json_buffer_echo")
            .and(warp::post())
            .and(warp::header::<String>("Authorization"))
            .and(warp::body::content_length_limit(1024 * 16))
            .and(warp::body::bytes())
            .and_then(handlers::create_json_buffer_echo)
    }

    pub fn avro_buffer() -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("avro_buffer")
            .and(warp::post())
            .and(warp::body::content_length_limit(1024 * 16))
            .and(warp::body::stream())
            .and_then(handlers::create_avro_buffer)
    }
    pub fn avro_buffer_echo() -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        let a = warp::path("a").and(warp::path("b")) ;
        warp::path!("sys" / String / "bucket" / String / "avro_buffer_echo")
            .and(warp::post())
            .and(warp::header::<String>("Authorization"))
            .and(warp::body::content_length_limit(1024 * 16))
            .and(warp::body::bytes())
            .and_then(handlers::create_avro_buffer_echo)
    }
}

mod models {
    use serde_derive::{Deserialize, Serialize} ;
    use serde_json::json ;

    #[derive(Deserialize, Serialize)]
    pub struct Response<T> {
        pub code: u8,                                                                                                          
        pub msg: String,                                                                                                       
        pub result: Option<T>                                                                                                  
    }

    #[derive(Deserialize, Serialize)]
    pub struct AvroBufferResult {
        pub size: u64,
    }
    #[derive(Deserialize, Serialize)]
    pub struct BufferEchoResult {
        pub content: String
    }

    #[derive(Deserialize, Serialize)]                                                                                                                               
    pub struct AccessToken {                                                                                                                                        
        pub start_ts: u64,                                                                                                                                          
        pub end_ts: u64,                                                                                                                                            
        pub ip: String,                                                                                                                                             
        pub payload_enc: String                                                                                                                                     
    }                                                                                                                                                               
}

mod models_impl {
    use super::models ;                                                                                                                                             
    use super::codecs ;                                                                                                                                             
    use super::utils ;                                                                                                                                              

    use std::collections::HashMap ;

    impl models::AccessToken {
        pub fn from_string(s: &str) -> Result<Self, String> {
            let access_token_m = s.split("&").map(|x| utils::tuple2(x, "=").unwrap()).collect::<HashMap<&str, &str>>() ;

            let start_ts_s = access_token_m.get("start_ts").ok_or("field start_ts not find!")? ;
            let end_ts_s = access_token_m.get("end_ts").ok_or("field end_ts not find!")? ;

            let start_ts = u64::from_str_radix(start_ts_s, 10)
                .map_err(|_| String::from("start_ts is not a valid numeric!"))? ;
            let end_ts = u64::from_str_radix(end_ts_s, 10)
                .map_err(|_| String::from("end_ts is not a valid numeric!"))? ;

            let ip = access_token_m.get("ip").ok_or("field ip not find!")?.to_string() ;
            let payload_enc = access_token_m.get("payload_enc").ok_or("field payload_enc not find!")?.to_string() ;
            Ok(models::AccessToken {start_ts, end_ts, ip, payload_enc})
        }
    }
}
mod handlers {
    use std::convert::Infallible;
    use std::io::Read ;

    use bytes ;
    use futures ;
    use futures::stream ;
    use futures_util::stream::StreamExt ;

    use warp::reply::json ;

    use flate2::bufread::GzDecoder;
    use flate2::write::GzEncoder;
    use flate2::Compression;

    use async_std::fs::File ;
    use async_std::io::ReadExt ;

    use super::models ;
    use super::codecs ;
    use super::utils ;
    
    pub async fn create_avro_buffer(stream : impl futures::Stream<Item = Result<impl bytes::Buf, warp::Error>>) -> Result<impl warp::Reply, Infallible> {
        // let mut file = File::create("test.txt").await.map(|_| String::from(""))? ;
        // let a = stream.take(5).collect::<bytes::Bytes>() ;
        
        // stream.for_each(move |chunk| { () ; }) ;

        let resp = match _create_avro_buffer(stream).await {
            Ok(resp) => resp,
            Err(msg) => models::Response { code: 1, msg, result: None }
        } ;
        Ok(json(&resp))
    }
    pub async fn _create_avro_buffer(stream : impl futures::Stream<Item = Result<impl bytes::Buf, warp::Error>>) -> Result<models::Response<models::AvroBufferResult>, String> {
        // let file = File::create("test.txt").await.map_err(|_| String::from(""))? ;
        let datas = stream.map(|x| {
            let b = x.unwrap(); 
            let x = b.bytes().to_vec() ;
            let y = std::str::from_utf8(&x).unwrap().to_string() ;
            println!("data:{}", y) ;
            y
            // let y = b.collect() ;
            // let z = b.iter() ;
        }).collect::<Vec<_>>().await ;

        println!("datas-size: {}", datas.len()) ;
        let avro_buffer_result = models::AvroBufferResult { size: 100 } ;
        Ok(models::Response { code: 0, msg: "success".to_string(), result: Some(avro_buffer_result) })
    }

    pub async fn create_json_buffer_echo(sys: String, bucket: String, authorization: String, bs: bytes::Bytes) -> Result<impl warp::Reply, Infallible> {
        let resp = match _create_json_buffer_echo(sys, bucket, authorization, bs).await {

            Ok(resp) => resp,
            Err(msg) => models::Response { code: 1, msg, result: None }
        } ;
        Ok(json(&resp))
    }

    pub async fn _create_json_buffer_echo(sys: String, bucket: String, authorization: String, bs_zip: bytes::Bytes) -> Result<models::Response<models::BufferEchoResult>, String> {
        verify_token(authorization)? ;

        let bv = codecs::gunzip(&bs_zip).map_err(|_| String::from("gunzip failed!"))? ;
        let bs = &bv[..] ;

        let json_val = serde_json::from_reader::<&[u8], serde_json::Value>(bs).unwrap() ;

        let content = format!("{:?}", json_val) ;
        let avro_buffer_echo_result = models::BufferEchoResult { content } ;
        Ok(models::Response { code: 0, msg: "success".to_string(), result: Some(avro_buffer_echo_result) })
    }

    pub async fn create_avro_buffer_echo(sys: String, bucket: String, authorization: String, bs: bytes::Bytes) -> Result<impl warp::Reply, Infallible> {
        let resp = match _create_avro_buffer_echo(sys, bucket, authorization, bs).await {
            Ok(resp) => resp,
            Err(msg) => models::Response { code: 1, msg, result: None }
        } ;
        Ok(json(&resp))
    }
    pub async fn _create_avro_buffer_echo(sys: String, bucket: String, authorization: String, bs_zip: bytes::Bytes) -> Result<models::Response<models::BufferEchoResult>, String> {
        println!("authorization:{}", authorization) ;
        verify_token(authorization)? ;
        // println!("bs:{}", codecs::base64(&bs_zip)) ;

        let bv = codecs::gunzip(&bs_zip).map_err(|_| String::from("gunzip failed!"))? ;
        let mut bs = &bv[..] ;

        let mut avro_file = File::open("../../data-model/schema/dc.avsc").await.map_err(|_| String::from("dc.avsc not found!"))? ;

        let mut schema_buf = String::new() ;
        avro_file.read_to_string(&mut schema_buf).await.map_err(|x| format!("avro file read error: {}", x.to_string())) ;

        let schema = codecs::avro_schema(schema_buf)? ;
        let avro_val = codecs::avro_from_bin(&schema, &mut bs)? ;
        let content = format!("{:?}", avro_val) ;
        let avro_buffer_echo_result = models::BufferEchoResult { content } ;
        Ok(models::Response { code: 0, msg: "success".to_string(), result: Some(avro_buffer_echo_result) })
    }
    pub fn verify_token(authorization: String) -> Result<bool, String> {
        let pubkey_b64 = String::from("3oJdJiBvUAbMCyX+Fw4+ttfvWhH7bKSrIpFDbvj+qz8=") ;
        let pubkey_bs = codecs::unbase64(&pubkey_b64)? ;
        let pubkey = codecs::pubkey(&pubkey_bs) ;

        let (token_type, signed_access_token) = utils::tuple2(&authorization, " ").ok_or("token format error: [<token_type> <token_val>!")? ;
        if token_type != "Bearer" { Err("Bearer token needed!")? ; }

        let (sign, access_token_s) = utils::tuple2(&signed_access_token, "&").ok_or("missing sign and data!")? ;                                              
        let (sign_name, sign_val) = utils::tuple2(&sign, "=").ok_or("key-value needed!")? ;                                                                         
        if sign_name != "sign" { Err("token sign is not first field!")? ; }                                                                                         


        let check = codecs::sign_verify(&pubkey, access_token_s, sign_val)? ;
        if ! check { Err("token sign error!")? ; }

        let access_token = models::AccessToken::from_string(access_token_s)? ;                                                                                                                                                            
        let now_ts = utils::now_ts() ;                                                                                                                                                                                                    
        println!("[debug] compare ts...") ;                                                                                                                                 println!("now-ts:{}, access_token.end_ts:{}", now_ts, access_token.end_ts) ;
        if access_token.end_ts < now_ts { Err("token expire!")? ;} 
        Ok(true) 
    }
}

mod codecs {
    use std::io::{Read, Write} ;

    use data_encoding::BASE64;

    use flate2::Compression;
    use flate2::write::GzEncoder;
    use flate2::read::GzDecoder;

    use avro_rs ;

    use ring::signature::{Signature, KeyPair, Ed25519KeyPair, ED25519, UnparsedPublicKey};

    use super::utils ;

    pub fn base64(s: &[u8]) -> String {
        BASE64.encode(s)
    }
    pub fn unbase64(s: &str) -> Result<Vec<u8>, String> {
        BASE64.decode(s.as_bytes()).map_err(|x| x.to_string())
    }
    pub fn gzip(bs: &[u8]) -> Result<Vec<u8>, std::io::Error> {
        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
        encoder.write_all(bs) ;
        encoder.finish()
    }
    pub fn gunzip(bs: &[u8]) ->  Result<Vec<u8>, std::io::Error> {
        let mut d = GzDecoder::new(bs);
        let mut v = Vec::new();
        d.read_to_end(&mut v)? ;
        Ok(v)
    }
    pub fn pubkey(pubkey: &[u8]) -> UnparsedPublicKey<&[u8]> {
        UnparsedPublicKey::new (&ED25519, &pubkey)
    }
    pub fn sign_verify(pubkey: &UnparsedPublicKey<&[u8]>, msg: &str, sign_b64: &str) -> Result<bool, String> {                                                      
        let sign = unbase64(sign_b64)? ;                                                                                                                            
        Ok(pubkey.verify(&msg.as_bytes(), &sign).is_ok())                                                                                                           
    }                                                                                                                                                               

    pub fn avro_schema(s: String) -> Result<avro_rs::Schema, String> {
        avro_rs::Schema::parse_str(&s).map_err(|_| String::from("avro schema parse error!"))
    }
    pub fn avro_from_bin(schema: &avro_rs::Schema, bs: &mut &[u8]) -> Result<avro_rs::types::Value, String> {
        avro_rs::from_avro_datum(schema, bs, None).map_err(|_| String::from("avro parse error!"))
    }
    pub fn avro_from_json(schema: &avro_rs:: Schema, json: &mut &[u8]) -> Result<avro_rs::types::Value, String> {
        unimplemented!()
    }
}

mod utils {
    use std::time::{SystemTime, UNIX_EPOCH} ;
    use std::collections::HashMap;

    use itertools::Itertools;

    pub fn now_ts() -> u64 {
        SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
    }

    pub fn tuple2<'a>(s: &'a str, sep: &str) -> Option<(&'a str, &'a str)> {
        s.splitn(2, sep).next_tuple::<(_, _)>()
    }
}
